package flink;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception{
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //获取kafka数据源
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.10:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"kafkauj");
        DataStreamSource<String> streamSource = env.addSource(new FlinkKafkaConsumer<String>("kafkauserjump", new SimpleStringSchema(), properties));

        //将数据格式转为json 提取时间戳生成Watermark
        SingleOutputStreamOperator<JSONObject> streamOperator = streamSource.map(JSONObject::parseObject).assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));

        //定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return value.getString("last_page_id") == null || value.getString("last_page_id").length() <= 0;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return value.getString("last_page_id") == null || value.getString("last_page_id").length() <= 0;
            }
        }).within(Time.seconds(10));

        //将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(streamOperator.keyBy(json -> json.getJSONObject("common").getString("mid")), pattern);

        //提取匹配上的和超时事件
        OutputTag<JSONObject> timeOutTag = new OutputTag<JSONObject>("timeOut") {
        };
        SingleOutputStreamOperator<JSONObject> outputStreamOperator = patternStream.select(timeOutTag, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp) throws Exception {
                return pattern.get("start").get(0);
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> pattern) throws Exception {
                return pattern.get("start").get(0);
            }
        });
        DataStream<JSONObject> timeOutDS = outputStreamOperator.getSideOutput(timeOutTag);

        //UNION两种事件
        DataStream<JSONObject> union = outputStreamOperator.union(timeOutDS);

        //将数据写入kafka
        union.print();
        union.addSink(new FlinkKafkaProducer<JSONObject>("192.168.1.10:9092", "userjumpkafka", new SerializationSchema<JSONObject>() {
            @Override
            public byte[] serialize(JSONObject element) {
                return element.toString().getBytes();
            }
        }));

        //启动程序
        env.execute("UserJumpDetailApp");
    }
}
